// Copyright (c) 2019 Presto Labs Pte. Ltd.
// Author: xguo

#pragma once

#include <algorithm>
#include <atomic>
#include <cctype>
#include <cmath>
#include <functional>
#include <memory>
#include <numeric>
#include <stdexcept>
#include <string>
#include <unordered_map>
#include <unordered_set>
#include <utility>
#include <vector>

#include <boost/circular_buffer.hpp>

#include "coin2/exchange/base/order/external.h"
#include "coin2/exchange/base/order/order_spec.h"
#include "coin2/exchange/base/order/risk_pending_order.h"
#include "coin2/exchange/base/order/symbol_cache.h"
#include "coin2/exchange/base/order/unknown_id_manager.h"

namespace coin2::exchange::base::order {

using OrderSubTopicId = coin::proto::OrderSubTopicId;

class OrderContextManager;

template <typename Enum>
typename std::enable_if<std::is_enum<Enum>::value, std::ostream>::type& operator<<(
    std::ostream& os,
    Enum value) {
  return os << static_cast<int>(value);
}

enum class SubmitStatus {
  UNKNOWN = 0,
  BEGIN = 1,
  SUBMITTED = 2,
  ACCEPTED = 3,
  REJECTED = 4,
  ERROR = 5,
};

enum class CancelStatus {
  UNKNOWN = 0,
  BEGIN = 1,
  SUBMITTED = 2,
  ACCEPTED = 3,
  REJECTED = 4,
  ERROR = 5,
  CONFIRMED = 6,
};

enum class FillStatus {
  UNKNOWN = 0,
  BEGIN = 1,
  NO_FILL = 2,
  PART_FILLED = 3,
  FULLY_FILLED = 4,
};

class SubmitManager {
 public:
  SubmitManager() = default;

  void HandleSubmitted(int64_t event_time) {
    CHECK_EQ(status_, SubmitStatus::BEGIN);
    submitted_time_ = event_time;
    status_ = SubmitStatus::SUBMITTED;
  }

  void HandleAccepted(int64_t event_time) {
    accepted_time_ = event_time;
    status_ = SubmitStatus::ACCEPTED;
  }

  void HandleRejected(int64_t event_time) {
    CHECK_EQ(status_, SubmitStatus::SUBMITTED);
    rejected_time_ = event_time;
    status_ = SubmitStatus::REJECTED;
  }

  void HandleError(int64_t event_time) {
    CHECK(status_ == SubmitStatus::SUBMITTED || status_ == SubmitStatus::ACCEPTED);
    error_time_ = event_time;
    status_ = SubmitStatus::ERROR;
  }

  SubmitStatus status() const { return status_; }
  int64_t submitted_time() const { return submitted_time_; }
  int64_t accepted_time() const { return accepted_time_; }
  int64_t rejected_time() const { return rejected_time_; }
  int64_t error_time() const { return error_time_; }

 private:
  SubmitStatus status_{SubmitStatus::BEGIN};
  int64_t submitted_time_{0};
  int64_t accepted_time_{0};
  int64_t rejected_time_{0};
  int64_t error_time_{0};
};

class SingleCancelManager {
 public:
  SingleCancelManager() = default;

  void HandleSubmitted(int64_t event_time) {
    CHECK_EQ(status_, CancelStatus::BEGIN);
    submitted_time_ = event_time;
    status_ = CancelStatus::SUBMITTED;
  }

  void HandleAccepted(int64_t event_time) {
    CHECK_EQ(status_, CancelStatus::SUBMITTED);
    accepted_time_ = event_time;
    status_ = CancelStatus::ACCEPTED;
  }

  void HandleRejected(int64_t event_time) {
    rejected_time_ = event_time;
    status_ = CancelStatus::REJECTED;
  }

  void HandleConfirmed(int64_t event_time, double canceled_qty) {
    CHECK_EQ(status_, CancelStatus::ACCEPTED);
    confirmed_time_ = event_time;
    canceled_qty_ = canceled_qty;
    status_ = CancelStatus::CONFIRMED;
  }

  void HandleAutoCanceled(int64_t event_time, double canceled_qty) {
    CHECK(status_ == CancelStatus::BEGIN || status_ == CancelStatus::REJECTED);
    confirmed_time_ = event_time;
    canceled_qty_ = canceled_qty;
    status_ = CancelStatus::CONFIRMED;
  }

  void HandleError(int64_t event_time) {
    CHECK(status_ == CancelStatus::SUBMITTED || status_ == CancelStatus::ACCEPTED);
    error_time_ = event_time;
    status_ = CancelStatus::ERROR;
  }

  void Reset() {
    status_ = CancelStatus::BEGIN;
    submitted_time_ = 0;
    accepted_time_ = 0;
    rejected_time_ = 0;
    error_time_ = 0;
    confirmed_time_ = 0;
    canceled_qty_ = 0.;
  }

  double canceled_qty() const { return canceled_qty_; }

  CancelStatus status() const { return status_; }
  int64_t submitted_time() const { return submitted_time_; }
  int64_t accepted_time() const { return accepted_time_; }
  int64_t rejected_time() const { return rejected_time_; }
  int64_t error_time() const { return error_time_; }
  int64_t confirmed_time() const { return confirmed_time_; }

 private:
  CancelStatus status_{CancelStatus::BEGIN};
  int64_t submitted_time_{0};
  int64_t accepted_time_{0};
  int64_t rejected_time_{0};
  int64_t error_time_{0};
  int64_t confirmed_time_{0};
  double canceled_qty_{0.};
};

class CancelManager {
 public:
  CancelManager() = default;

  void HandleSubmitted(int64_t event_time) {
    switch (status()) {
      case CancelStatus::BEGIN:
        GetCurrManager().HandleSubmitted(event_time);
        break;
      default: {
      };
    }
  }

  void HandleAccepted(int64_t event_time) {
    switch (status()) {
      case CancelStatus::SUBMITTED:
        GetCurrManager().HandleAccepted(event_time);
        break;
      case CancelStatus::ACCEPTED:
      case CancelStatus::CONFIRMED:
        break;
      default:
        NOTREACHED() << status();
    }
  }

  void HandleRejected(int64_t event_time) {
    switch (status()) {
      // surprisingly, this is possible.
      // and valid.
      case CancelStatus::SUBMITTED:
      // in 2020-07-27, it happened.
      // please see ec2-user@strategy-57.ap-northeast-1.huobi:/home/ec2-user/sung.log
      case CancelStatus::ACCEPTED:
      case CancelStatus::CONFIRMED:
        GetCurrManager().HandleRejected(event_time);
        Inc();
        break;
      case CancelStatus::BEGIN:
      case CancelStatus::REJECTED:
        break;
      default:
        NOTREACHED() << status();
    }
  }

  void HandleError(int64_t event_time) {
    switch (status()) {
      case CancelStatus::SUBMITTED:
        GetCurrManager().HandleError(event_time);
        Inc();
        break;
      case CancelStatus::ERROR:
        break;
      default:
        NOTREACHED() << status();
    }
  }

  void HandleConfirmed(int64_t event_time, double canceled_qty) {
    switch (status()) {
      // if cancel timeout, get rejected, then get cancel response, it's Begin
      case CancelStatus::BEGIN:
        HandleAutoCanceled(event_time, canceled_qty);
        break;
      case CancelStatus::SUBMITTED:
      case CancelStatus::ACCEPTED:
        GetCurrManager().HandleConfirmed(event_time, canceled_qty);
        canceled_qty_ += canceled_qty;
        break;
      case CancelStatus::CONFIRMED:
        break;
      default:
        NOTREACHED() << status();
    }
  }

  void HandleAutoCanceled(int64_t event_time, double canceled_qty) {
    switch (status()) {
      case CancelStatus::BEGIN:
      case CancelStatus::REJECTED:
        GetCurrManager().HandleAutoCanceled(event_time, canceled_qty);
        canceled_qty_ += canceled_qty;
        break;
      default: {
      };
    }
  }

  int num_of_cancels() const { return current_idx_ + 1; }

  double canceled_qty() const { return canceled_qty_; }

  bool IsCancelSent() const {
    const auto status = GetCurrManager().status();
    return status == CancelStatus::SUBMITTED || status == CancelStatus::ACCEPTED;
  }

  CancelStatus status() const { return GetCurrManager().status(); }

  CancelStatus last_status() const {
    CHECK(current_idx_ >= 1) << current_idx_;
    return GetPrevManager().status();
  }

  int64_t last_cancel_sent_time() const { return GetPrevManager().submitted_time(); }

  int64_t submitted_time() const {
    if (current_idx_ >= kMaxNumCancels) {
      LOG(WARNING) << "Too many cancels: " << current_idx_ << ", id: " << this;
    }
    return GetCurrManager().submitted_time();
  }
  int64_t accepted_time() const { return GetCurrManager().accepted_time(); }
  int64_t rejected_time() const { return GetCurrManager().rejected_time(); }
  int64_t error_time() const { return GetCurrManager().error_time(); }
  int64_t confirmed_time() const { return GetCurrManager().confirmed_time(); }

 private:
  SingleCancelManager& GetCurrManager() { return single_cancel_managers_[current_idx_ % 2]; }
  const SingleCancelManager& GetCurrManager() const {
    return single_cancel_managers_[current_idx_ % 2];
  }
  const SingleCancelManager& GetPrevManager() const {
    return single_cancel_managers_[(current_idx_ + 1) % 2];
  }
  void Inc() {
    current_idx_++;
    GetCurrManager().Reset();
  }

 private:
  SingleCancelManager single_cancel_managers_[2];
  double canceled_qty_{0};
  int current_idx_{0};
  static const int kMaxNumCancels = 10;
};

struct FillElement {
  int64_t timestamp;
  coin::proto::OrderSide side;
  double price;
  double qty;
  std::string fill_id;
  coin::proto::FillType fill_type;
};

class FillManager {
 public:
  FillManager(double price_orig, double qty_orig)
      : price_orig_(price_orig),
        qty_orig_(qty_orig),
        price_avg_fill_(0.0),
        qty_accum_fill_(0.0),
        last_fill_(nullptr),
        status_(FillStatus::BEGIN) {}

  FillManager(const FillManager&) = delete;

  bool IsFullyFilled() const;
  bool HasFill(const std::string& fill_id) const { return fills_.count(fill_id) > 0; }

  FillElement* GetMutableFill(const std::string& fill_id) {
    auto it = fills_.find(fill_id);
    if (it != fills_.end()) {
      return it->second.get();
    } else {
      return nullptr;
    }
  }

  bool UpdateByLastFill(
      coin::proto::OrderSide side,
      double price_last_fill,
      double qty_last_fill,
      const std::string& fill_id,
      coin::proto::FillType fill_type,
      int64_t timestamp) {
    if (!IsReasonableFilledQty(qty_accum_fill_ + qty_last_fill, qty_orig_)) {
      LOG(WARNING) << fmt::format(
          "Insane qty_accum_fill_ (update_by_last_fill) qty_orig={:.3f}, qty_accum_fill={:.3f}",
          qty_orig_,
          (qty_accum_fill_ + qty_last_fill));
      return false;
    }
    price_avg_fill_ = (price_avg_fill_ * qty_accum_fill_ + price_last_fill * qty_last_fill) /
                      (qty_accum_fill_ + qty_last_fill);
    qty_accum_fill_ += qty_last_fill;
    fills_[fill_id].reset(
        new FillElement{timestamp, side, price_last_fill, qty_last_fill, fill_id, fill_type});
    last_fill_ = fills_[fill_id].get();
    if (IsFullyFilled()) {
      status_ = FillStatus::FULLY_FILLED;
    } else {
      status_ = FillStatus::PART_FILLED;
    }
    return true;
  }

  size_t GetFillCount() const { return fills_.size(); }
  double price_orig() const { return price_orig_; }
  double qty_orig() const { return qty_orig_; }
  double price_avg_fill() const { return price_avg_fill_; }
  double qty_accum_fill() const { return qty_accum_fill_; }
  const FillElement* last_fill() const { return last_fill_; }
  FillStatus status() { return status_; }
  double filled_qty() const { return qty_accum_fill_; }
  auto GetFills() const -> std::vector<const FillElement*> {
    std::vector<const FillElement*> ret;
    for (const auto& [id, elem] : fills_) {
      ret.push_back(elem.get());
    }
    return ret;
  }

 private:
  static bool IsReasonableFilledQty(double filled_qty, double orig_qty);

 private:
  double price_orig_;
  double qty_orig_;
  double price_avg_fill_;
  double qty_accum_fill_;
  const FillElement* last_fill_;
  std::unordered_map<std::string, std::unique_ptr<FillElement>> fills_;
  FillStatus status_;
};

class OrderContext {
 public:
  using OrderSimpleStatus = coin::proto::OrderState;
  using OrderSpec = coin2::exchange::base::order::OrderSpec;
  using IProduct = coin2::exchange::base::symbology::IProduct;
  using OrderEventHandler = std::function<void(const coin::proto::OrderEvent&)>;
  using OrderEventHandlerList = std::vector<OrderEventHandler>;

 public:
  OrderContext(int64_t proc_order_id, const OrderSpec& order_spec)
      : proc_order_id_(proc_order_id),
        order_spec_(order_spec),
        last_seen_time_(order_spec.order_created_time),
        last_time_query_alive_(order_spec.order_created_time),
        status_(OrderSimpleStatus::CREATED_ORDER),
        fill_manager_(order_spec.price, order_spec.qty),
        is_foreign_(false),
        handler_list_(nullptr),
        latency_context_id_(0) {}
  OrderContext(const OrderContext&) = delete;

 public:
  const OrderSpec& order_spec() const { return order_spec_; }
  int64_t proc_order_id() const { return proc_order_id_; }
  const std::string& exchange_order_id() const { return exchange_order_id_; }
  bool has_exchange_order_id() const { return !exchange_order_id_.empty(); }
  OrderSimpleStatus order_state() const { return status_; }
  OrderSimpleStatus status() const { return status_; }
  bool implied_fill_published() const { return implied_fill_published_; }

  void SetExchangeOrderId(const std::string& exchange_order_id) {
    exchange_order_id_ = exchange_order_id;
  }

  void SetForeign() { is_foreign_ = true; }

  bool has_fill_id(const std::string& fill_id) { return fill_manager_.HasFill(fill_id); }

  double filled_qty() const { return fill_manager_.filled_qty(); }
  size_t filled_cnt() const { return fill_manager_.GetFillCount(); }

  double position() const { return filled_qty() * order_spec_.sign(); }

  double working_qty() const {
    return (order_spec_.qty - cancel_manager_.canceled_qty() - fill_manager_.filled_qty());
  }

  const std::string& native_product() const { return order_spec_.product; }

  OrderSide order_side() const { return order_spec_.order_side; }

  OrderDirection direction() const { return order_spec_.direction(); }

  double order_price() const { return order_spec_.price; }

  double order_qty() const { return order_spec_.qty; }

  OrderSide last_fill_side() const { return fill_manager_.last_fill()->side; }
  double last_fill_qty() const { return fill_manager_.last_fill()->qty; }

  const SubmitManager& submit_manager() const { return submit_manager_; }
  const CancelManager& cancel_manager() const { return cancel_manager_; }
  const FillManager& fill_manager() const { return fill_manager_; }

  const std::string& tag() const { return order_spec_.tag; }
  bool is_accepted() const { return has_exchange_order_id(); }
  bool is_alive() const { return status_ != OrderSimpleStatus::DEAD_ORDER; }
  bool is_permanent() const { return is_permanent_; }
  void set_permanent() { is_permanent_ = true; }

  bool IsForeign() const { return is_foreign_; }
  bool IsCancelSent() const { return cancel_manager_.IsCancelSent(); }
  bool IsCancelAccepted() const { return cancel_manager_.status() == CancelStatus::ACCEPTED; }
  bool IsCancelConfirmed() const { return cancel_manager_.status() == CancelStatus::CONFIRMED; }
  bool IsFullyFilled() const { return fill_manager_.IsFullyFilled(); }

  int64_t order_created_time() const { return order_spec().order_created_time; }
  int64_t order_accepted_time() const { return submit_manager().accepted_time(); }
  int64_t cancel_created_time() const { return cancel_manager().submitted_time(); }
  int64_t cancel_confirmed_time() const { return cancel_manager().confirmed_time(); }
  bool is_cancel_pending() const { return IsCancelSent(); }

  int64_t last_seen_time() const { return last_seen_time_; }
  int64_t last_time_query_alive() const { return last_time_query_alive_; }
  int64_t GetOrderAge(int64_t curr_ts) const { return curr_ts - order_spec().order_created_time; }

  void RegisterOrderEventHandlers(OrderEventHandlerList* handler_list) {
    handler_list_ = handler_list;
  }

  void UpdateLastSeenTime(int64_t event_time) { last_seen_time_ = event_time; }
  void UpdateLastTimeQueryAlive(int64_t event_time) { last_time_query_alive_ = event_time; }

  void HandleOrderSubmitted(int64_t event_time);
  void HandleOrderAccepted(int64_t event_time, const std::string& exchange_order_id);
  void HandleOrderRejected(
      int64_t event_time,
      coin::proto::OrderErrorCode error_code,
      const std::string& exchange_order_code,
      const std::string& exchange_order_msg);
  void HandleOrderError(int64_t event_time);
  void HandleCancelSubmitted(int64_t event_time);
  void HandleCancelAccepted(int64_t event_time);
  void HandleCancelRejected(
      int64_t event_time,
      coin::proto::OrderErrorCode error_code,
      const std::string& exchange_order_code,
      const std::string& exchange_order_msg);
  void HandleCancelError(int64_t event_time);
  void HandleCancelConfirmed(int64_t event_time, double canceled_qty);
  void HandleCancelConfirmed(int64_t event_time);
  void HandleOrderAutoCanceled(int64_t event_time, double canceled_qty);
  void HandleFill(
      int64_t event_time,
      double filled_price,
      double filled_qty,
      const std::string& fill_id,
      coin::proto::FillType fill_type,
      coin::proto::FillSource fill_source,
      std::optional<double> fill_fee,
      std::optional<std::string> fill_fee_currency);
  void HandleProcIdReplaced(int64_t event_time, const OrderContext& oc_alt);
  void HandleUnknownOrder(int64_t event_time);

  coin::proto::OrderEvent CreateOrderEvent();
  void PublishOrderEvent(coin::proto::OrderEvent&& oe);

  int64_t latency_context_id() const { return latency_context_id_; }
  void set_latency_context_id(int64_t latency_context_id) const {
    latency_context_id_ = latency_context_id;
  }

  OrderContextProto ToProto() const {
    OrderContextProto proto_msg;

    *proto_msg.mutable_order_spec() = order_spec_.ToProto();
    proto_msg.set_proc_order_id(proc_order_id_);
    if (!exchange_order_id_.empty()) {
      proto_msg.set_exchange_order_id(exchange_order_id_);
    }
    proto_msg.set_dead_timestamp(dead_timestamp_);
    proto_msg.set_dead_reason_code(dead_reason_code_);

    return proto_msg;
  }

  static std::unique_ptr<OrderContext> FromProto(const OrderContextProto& proto) {
    auto order_spec = OrderSpec::FromProto(proto.order_spec());
    auto order_context = std::make_unique<OrderContext>(proto.proc_order_id(), *order_spec);
    if (proto.has_exchange_order_id()) {
      order_context->SetExchangeOrderId(proto.exchange_order_id());
    }
    if (proto.has_dead_timestamp()) {
      order_context->dead_timestamp(proto.dead_timestamp());
    }
    if (proto.has_dead_reason_code()) {
      order_context->dead_reason_code(proto.dead_reason_code());
    }

    return order_context;
  }

  int num_of_cancels() const { return cancel_manager_.num_of_cancels(); }

  auto dead_timestamp() -> int64_t { return dead_timestamp_; }
  void dead_timestamp(int64_t timestamp) { dead_timestamp_ = timestamp; }
  auto dead_reason_code() -> coin::proto::OrderErrorCode { return dead_reason_code_; }
  void dead_reason_code(coin::proto::OrderErrorCode code) { dead_reason_code_ = code; }

  bool IsAmendOrder() const { return order_spec_.use_amend; }
  auto GetAmendedCnt() const -> int32_t {
    // it's not accurate in some cases
    return order_replaced_count_;
  }
  auto SetAmendedCnt(int32_t amended_cnt) { order_replaced_count_ = amended_cnt; }

  bool IsClosible() const;

  void RegisterOrderContextManager(OrderContextManager* ocm) { ocm_ = ocm; }

  // for amend order
  void ReplacedByOrder(int64_t timestamp, OrderContext* oc);
  void UpdateByLastFill(
      int64_t event_time,
      coin::proto::OrderSide order_side,
      double price_last_fill,
      double qty_last_fill,
      const std::string& fill_id,
      coin::proto::FillType fill_type);

 private:
  const int64_t proc_order_id_;
  const OrderSpec order_spec_;
  std::string exchange_order_id_;
  int64_t last_seen_time_;
  int64_t last_time_query_alive_;
  bool is_permanent_{false};  // for bithumb only.

  OrderSimpleStatus status_;

  SubmitManager submit_manager_;
  CancelManager cancel_manager_;
  FillManager fill_manager_;
  bool is_foreign_{false};
  bool implied_fill_published_{false};
  OrderEventHandlerList* handler_list_;
  mutable int64_t latency_context_id_;

  // for amend order
  int64_t replaced_proc_order_id_ = 0;
  int32_t order_replaced_count_ = 0;

  // for abnormal case
  int64_t dead_timestamp_ = 0;
  coin::proto::OrderErrorCode dead_reason_code_ = coin::proto::OrderErrorCode::ORDER_UNKNOWN_ERROR;

  OrderContextManager* ocm_ = nullptr;
};

struct OrderContextHolder {
  OrderContextHolder(std::unique_ptr<OrderContext>&& order_context, OrderContextManager* ocm)
      : order_context(std::move(order_context)), ocm(ocm) {}

  ~OrderContextHolder();
  std::unique_ptr<OrderContext> order_context;
  OrderContextManager* ocm;
};

struct OrderContextUpdateResult {
  coin::proto::OrderEvent order_event;
  nlohmann::json extra_info;

  bool has_debug_info() const;
  void Clear();
  void AddInfo(const std::string& location, const nlohmann::json& info);
  void CreateDebugOrderEvent();
};

class OrderContextManager {
 public:
  using OrderSimpleStatus = coin::proto::OrderState;
  using OrderContextManagerSummaryProto = coin::proto::OrderContextManagerSummaryProto;
  using OrderEvent = coin::proto::OrderEvent;
  using ProductOrderBundle = coin::proto::ProductOrderBundle;
  using ProductOrderElement = coin::proto::ProductOrderElement;
  using ProductFillBundle = coin::proto::ProductFillBundle;
  using ProductFillElement = coin::proto::ProductFillElement;
  using ConversionBundle = coin::proto::ConversionBundle;
  using ConversionElement = coin::proto::ConversionElement;
  using SubmitOrderResponse = coin::proto::SubmitOrderResponse;
  using MultiSubmitResponse = coin::proto::MultiSubmitResponse;
  using CancelOrderResponse = coin::proto::CancelOrderResponse;
  using MultiCancelResponse = coin::proto::MultiCancelResponse;
  using SubmitOrderResponseBundle = coin::proto::SubmitOrderResponseBundle;
  using CancelOrderResponseBundle = coin::proto::CancelOrderResponseBundle;
  using OrderContextContainer = boost::circular_buffer<std::unique_ptr<OrderContextHolder>>;
  using OrderSpec = order::OrderSpec;
  using RiskEventHandler = coin2::exchange::base::order::RiskPendingOrder::RiskEventHandler;

 public:
  class OrderEventPublisher {
   public:
    OrderEventPublisher() = default;
    ~OrderEventPublisher() = default;

    void RegisterOrderEventHandlers(OrderContext::OrderEventHandlerList* handler_list) {
      handler_list_ = handler_list;
    }
    void PublishOrderEvent(coin::proto::OrderEvent&& oe, bool flush = false);
    void Reset();
    void Flush();

   private:
    OrderContext::OrderEventHandlerList* handler_list_ = nullptr;
    std::atomic<bool> publish_in_progress_ = false;
    std::vector<coin::proto::OrderEvent> oe_to_publish_pending_ = {};
    std::vector<coin::proto::OrderEvent> oe_to_publish_ongoing_ = {};
  };

  class OrderEventFlusher {
   public:
    explicit OrderEventFlusher(OrderEventPublisher* order_event_publisher)
        : order_event_publisher_{order_event_publisher} {
      if (order_event_publisher_) {
        order_event_publisher_->Reset();
      }
    }

    ~OrderEventFlusher() {
      if (order_event_publisher_) {
        order_event_publisher_->Flush();
      }
    }

   private:
    OrderEventPublisher* order_event_publisher_ = nullptr;
  };

 public:
  explicit OrderContextManager(const MarketExchangeApi& mea, int64_t queue_size = 4096 * 16)
      : mea_(mea), order_list_(queue_size) {
    SetDeadOrderAliveRestartSec(mea);
    SetBookFeedFillInterval(mea);
    order_event_publisher_ = std::make_unique<OrderEventPublisher>();
    risk_pending_order_ = std::make_unique<RiskPendingOrder>(30, 900 * 1'000'000'000LL);
    system_start_ts_ = GetCurrentTimestamp();
  }

  auto GenerateOrderEventFlusher() -> std::unique_ptr<OrderEventFlusher> {
    return std::make_unique<OrderEventFlusher>(order_event_publisher_.get());
  }
  void PublishOrderEvent(coin::proto::OrderEvent&& oe, bool flush = true);
  void Reset();
  void Flush();

  void DeleteOrderMap(const OrderContext& order_context) {
    // 1. erase proc_order_id map
    const auto& proc_order_id = order_context.proc_order_id();
    proc_order_id_map_.erase(proc_order_id);
    // 2. erase exchange_order_id map
    const auto& exchange_order_id = order_context.exchange_order_id();
    auto iter1 = exchange_order_id_map_.find(exchange_order_id);
    if (iter1 != exchange_order_id_map_.end() && proc_order_id == iter1->second->proc_order_id()) {
      exchange_order_id_map_.erase(iter1);
    }
    // 3. erase multi_orders map
    const auto key = std::make_pair(exchange_order_id, proc_order_id);
    auto iter2 = multi_orders_map_.find(key);
    if (iter2 != multi_orders_map_.end()) {
      multi_orders_map_.erase(iter2);
    }
  }

  ~OrderContextManager() = default;

  const MarketExchangeApi& mea() const { return mea_; }

  std::size_t size() const { return order_list_.size(); }
  std::size_t live_orders_size() const { return live_orders_.size(); }
  int32_t GetWorkingOrderCount() const { return static_cast<int32_t>(live_orders_size()); }

  void SetDead(int64_t proc_order_id);
  void SetDeadWithErrorCode(int64_t proc_order_id, coin::proto::OrderErrorCode error_code);
  std::unordered_map<int64_t, OrderContext*>::iterator SetDeadIterator(
      std::unordered_map<int64_t, OrderContext*>::iterator it,
      coin::proto::OrderErrorCode error_code);
  void AddBackWorkingOrders(OrderContext*);
  bool HasProcOrderId(int64_t proc_order_id) const;
  bool HasExchangeOrderId(const std::string& exchange_order_id) const;
  bool HasOrder(const OrderContext& order_context) const;

  auto FindByProcOrderId(int64_t proc_order_id) -> OrderContext*;
  auto FindByProcOrderId(const int64_t proc_order_id) const -> const OrderContext*;
  auto FindByExchangeOrderId(const std::string& exchange_order_id) -> OrderContext*;
  auto FindSubmitPending(const coin::proto::ProductOrderElement&) -> OrderContext*;
  auto FindByOrderIds(
      const std::string& symbol,
      const std::string& exchange_order_id,
      int64_t proc_order_id) -> OrderContext*;
  auto IterateSearchByOrderIds(const std::string& exchange_order_id, int64_t proc_order_id)
      -> OrderContext*;

  OrderContextManagerSummaryProto GetLiveOrderSummary() const;
  std::string DebugString(std::optional<int64_t> timestamp = {}) const;

  OrderContext* AddOrderContext(std::unique_ptr<OrderContext>&& order_context);
  OrderContext* AddOrderContext(const OrderSpec& order_spec);
  OrderContext* AddOrderContext(const ProductOrderElement& product_order_element);
  bool PostValidateOrderContext(const OrderContext* order_context) const;

  const std::unordered_set<std::string>& native_symbols() const { return native_symbols_; }

  void SetExchangeOrderId(OrderContext* oc, const std::string& exchange_order_id);
  bool IsManagedSymbol(const std::string& native_symbol) const {
    return native_symbols_.count(native_symbol) > 0;
  }

  void ValidateConsistency();

  int64_t AllocateProcOrderId() const { return proc_order_id_; }
  int64_t AllocateProcOrderIdAndInc() {
    auto ret = proc_order_id_;
    proc_order_id_++;

    return ret;
  }
  void HandleBookFeed(const IProduct& product, double ask0, double bid0);
  void HandleOrderExecution(const OrderExecutionProto& order_execution);
  void HandleOrderInfoList(const ProductOrderBundle& product_order_bundle);
  void HandleFillInfoList(const ProductFillBundle& product_fill_bundle);
  void HandleConversionList(const ConversionBundle& conversion_bundle);
  void HandleSubmitResponse(const SubmitOrderResponse& submit_order_response);
  void HandleSubmitBatchResponse(const MultiSubmitResponse& each_submit_response);
  void HandleCancelResponse(const CancelOrderResponse& cancel_order_response);
  void HandleCancelBatchResponse(const MultiCancelResponse& each_cancel_response);
  void HandleAutoCancelAllProduct(int64_t ts) {
    auto order_event_publish_helper = GenerateOrderEventFlusher();
    auto tmp_live_orders = live_orders_;
    for (auto each : tmp_live_orders) {
      auto oc = each.second;
      if (oc->cancel_manager().status() == CancelStatus::BEGIN) {
        oc->HandleOrderAutoCanceled(ts, oc->working_qty());
      } else {
        if (!oc->IsCancelAccepted()) {
          oc->HandleCancelAccepted(ts);
        }
        oc->HandleCancelConfirmed(ts, oc->working_qty());
      }
      SetDead(oc->proc_order_id());
    }
  }
  void HandleAmendOrderAccepted(int64_t timestamp, OrderContext* oc);

  void RegisterOrderEventHandler(const OrderContext::OrderEventHandler& handler);
  void RegisterRiskEventHandler(const RiskEventHandler& handler);

  bool IsInWorkingOrders(const OrderContext* oc) const {
    return live_orders_.find(oc->proc_order_id()) != live_orders_.end();
  }
  const std::vector<const OrderContext*>& GetAllWorkingOrders() const {
    return_orders_list_.clear();
    for (const auto& [poid, oc] : live_orders_) {
      (void)poid;
      if (oc->is_alive()) {
        return_orders_list_.emplace_back(oc);
      }
    }
    return return_orders_list_;
  }
  const std::vector<const OrderContext*>& GetAllProductWorkingOrders(
      const std::string& native_symbol) const {
    return_orders_list_.clear();
    if (product_live_orders_.count(native_symbol) > 0) {
      for (const auto& [poid, oc] : product_live_orders_.at(native_symbol)) {
        (void)poid;
        if (oc->is_alive()) {
          return_orders_list_.emplace_back(oc);
        }
      }
    }
    return return_orders_list_;
  }

  double GetFillNotSyncedToBuyPos(const std::string& native_symbol) const {
    if (fill_not_synced_to_buy_pos_.count(native_symbol) == 0) {
      return 0;
    }
    return fill_not_synced_to_buy_pos_.at(native_symbol);
  }
  double GetFillNotSyncedToSellPos(const std::string& native_symbol) const {
    if (fill_not_synced_to_sell_pos_.count(native_symbol) == 0) {
      return 0;
    }
    return fill_not_synced_to_sell_pos_.at(native_symbol);
  }
  void ClearFillNotSyncedPos(const std::string& native_symbol) {
    fill_not_synced_to_buy_pos_[native_symbol] = 0.0;
    fill_not_synced_to_sell_pos_[native_symbol] = 0.0;
  }
  void ClearFillNotSyncedPosFromCurrency(const std::string& native_currency) {
    for (const auto& norm_product : norm_products_) {
      if (norm_product->base().native_currency() == native_currency) {
        ClearFillNotSyncedPos(norm_product->order_native_symbol());
      }
    }
  }
  void RemoveTimeoutPendingOrders(int64_t timestamp);
  void RemoveZombieOrders(int64_t timestamp, int64_t max_order_idle_age);
  std::vector<int64_t> GetZombieOrders(int64_t timestamp, int64_t max_age);
  std::vector<int64_t> GetCancelRejectedOrders();
  void SetTimeoutCancelRejected(const OrderContext* order_context, int64_t timestamp);
  void ManageRisk(int64_t timestamp);

  void SetSymcache(const SymbologyCache* symcache) {
    symcache_ = symcache;
    for (auto product : symcache_->GetProducts()) {
      native_symbols_.emplace(product->native_symbol());
      ClearFillNotSyncedPos(product->native_symbol());
      norm_products_.emplace_back(product);
    }
  }

  void UpdateTimestamp(int64_t ts) {
    // First time current_ts gets updated.
    if (proc_order_id_ == 0) {
      proc_order_id_ = (ts / 1'000'000'000LL) * 1'000'000'000LL +
                       (reinterpret_cast<int64_t>(this) % 1'000'000'000LL);
    }

    if (ts >= current_ts_) {
      current_ts_ = ts;
    } else {
      LOG(ERROR) << "reversed time: " << current_ts_ << ", " << ts;
    }
  }

  int64_t get_ts() const {
    CHECK_GT(current_ts_, 10000);
    return current_ts_;
  }

  const auto& result() const { return result_; }

  void ClearResult() {
    result_.Clear();
    result_.order_event.set_event_time(current_ts_);
  }

  void SetSubmitPendingTimeout(int64_t submit_pending_timeout_sec) {
    submit_pending_timeout_ = submit_pending_timeout_sec * NS;
  }

 private:
  void HandleSubmitResponseWithoutFlush(const SubmitOrderResponse& submit_order_response);
  void HandleCancelResponseWithoutFlush(const CancelOrderResponse& cancel_order_response);
  void SetDeadOrderAliveRestartSec(const MarketExchangeApi& mea);
  void SetBookFeedFillInterval(const MarketExchangeApi& mea);
  bool IsTickAligned(const OrderSpec& order_spec) const;

 private:
  const int64_t NS = 1'000'000'000LL;
  const MarketExchangeApi mea_;
  int64_t submit_pending_timeout_{30 * NS};
  int64_t proc_order_id_{0};

  std::unordered_map<int64_t, OrderContext*> proc_order_id_map_;
  std::unordered_map<std::string, OrderContext*> exchange_order_id_map_;
  typedef std::pair<std::string, int64_t> MultiOrdersKeyType;
  std::map<MultiOrdersKeyType, OrderContext*> multi_orders_map_;  // TODO(daniel): make clear name
  // note: need to be destroyed later than above maps
  OrderContextContainer order_list_;

  // note: please update product_live_orders_ along with live_orders_
  std::unordered_map<int64_t, OrderContext*> live_orders_;
  std::unordered_map<std::string, std::unordered_map<int64_t, OrderContext*>> product_live_orders_;
  const std::unordered_map<int64_t, OrderContext*> empty_orders_;
  OrderContext::OrderEventHandlerList handler_list_;
  std::unordered_set<std::string> native_symbols_;
  std::vector<const base::symbology::IProduct*> norm_products_;

  UnknownIdManager unknown_fill_id_manager_;
  UnknownIdManager unknown_order_id_manager_;

  // conversion
  std::unordered_set<std::string> conversion_set_;

  std::unordered_map<std::string, double> fill_not_synced_to_buy_pos_;
  std::unordered_map<std::string, double> fill_not_synced_to_sell_pos_;

  const SymbologyCache* symcache_{nullptr};

  mutable std::vector<const OrderContext*> return_orders_list_;

  int dead_order_alive_restart_sec_;
  std::unordered_map<int64_t, int> dead_order_alive_times_map_;

  int64_t book_feed_fill_interval_;
  int64_t current_ts_{0};

  OrderContextUpdateResult result_;

  std::unique_ptr<OrderEventPublisher> order_event_publisher_ = {};
  std::unique_ptr<RiskPendingOrder> risk_pending_order_ = {};

  int64_t system_start_ts_ = 0;
};

using IOrderContextManager = OrderContextManager;
using IOrderContext = OrderContext;

}  // namespace coin2::exchange::base::order
